package etl

import beans.Info
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.{SparkContext, SparkConf}

/**
  * Created by 王康 on 2018/6/27.
  */
object Bz2ParquteV2 {
  def main(args: Array[String]) {
    //设置配置文件
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName).setMaster("local[4]")
    //创建spark连接
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val scc: SparkContext = new SparkContext(conf)
    //创建sparksql 连接
    val sqlcontext: SQLContext = new SQLContext(scc)
    //在sqlcontext 对象上设置  以 snappy格式来压缩文件
    sqlcontext.setConf("spark.sql.parquet.compression.codec", "snappy")
    //导入隐式转换
    import sqlcontext.implicits._
    val file: RDD[String] = scc.textFile("E:\\大数据资料\\project2\\dmpbig.FINISH")

    val df: DataFrame = file.map(t => t.split(",")).filter(t => t.length == 85 && !t(20).isEmpty).map(sche => {

      new Info(sche(0),
        sche(1).toInt,
        sche(2).toInt,
        sche(3).toInt,
        sche(4).toInt,
        sche(5),
        sche(6),
        sche(7).toInt,
        sche(8).toInt,
        sche(9).toInt,
        sche(10).toDouble,
        sche(11),
        sche(12),
        sche(13),
        sche(14),
        sche(15),
        sche(16),
        sche(17).toInt,
        sche(18),
        sche(19),
        sche(20).toInt,
        sche(21).toInt,
        sche(22),
        sche(23),
        sche(24),
        sche(25),
        sche(26).toInt,
        sche(27),
        sche(28).toInt,
        sche(29),
        sche(30).toInt,
        sche(31).toInt,
        sche(32).toInt,
        sche(33),
        sche(34).toInt,
        sche(35).toInt,
        sche(36).toInt,
        sche(37),
        sche(38).toInt,
        sche(39).toInt,
        sche(40).toDouble,
        sche(41).toDouble,
        sche(42).toInt,
        sche(43),
        sche(44).toDouble,
        sche(45).toDouble,
        sche(46),
        sche(47),
        sche(48),
        sche(49),
        sche(50),
        sche(51),
        sche(52),
        sche(53),
        sche(54),
        sche(55),
        sche(56),
        sche(57).toInt,
        sche(58).toDouble,
        sche(59).toInt,
        sche(60).toInt,
        sche(61),
        sche(62),
        sche(63),
        sche(64),
        sche(65),
        sche(66),
        sche(67),
        sche(68),
        sche(69),
        sche(70),
        sche(71),
        sche(72),
        sche(73).toInt,
        sche(74).toDouble,
        sche(75).toDouble,
        sche(76).toDouble,
        sche(77).toDouble,
        sche(78).toDouble,
        sche(79),
        sche(80),
        sche(81),
        sche(82),
        sche(83),
        sche(84).toInt)
    }).toDF()
    df.write.parquet("E:\\大数据资料\\project2\\parquet_1.6.3test")

    df.printSchema()

    scc.stop()


  }
}


